package com.gitee.ixtf.vertx.verticle

import com.gitee.ixtf.vertx.kotlinx.failure
import io.vertx.core.eventbus.Message
import io.vertx.kotlin.coroutines.awaitResult
import io.vertx.kotlin.coroutines.coAwait
import io.vertx.kotlin.coroutines.toReceiveChannel
import kotlinx.coroutines.channels.consumeEach
import kotlinx.coroutines.launch

abstract class ConsumerCoroutineVerticle<T, R>(address: String) : BaseCoroutineVerticle() {
  private val consumer by lazy { eventBus.consumer<T>(address) }
  protected open val parallel = true

  protected abstract suspend fun handle(reply: Message<T>): R

  protected open suspend fun handleSuccess(reply: Message<T>, r: R) = reply.reply(r)

  protected open suspend fun handleFailure(reply: Message<T>, t: Throwable) = reply.failure(log, t)

  protected open suspend fun handleFinally(reply: Message<T>) = Unit

  override suspend fun start() {
    if (parallel) consumer.coHandler { consume(it) }
    else launch { consumer.toReceiveChannel(vertx).consumeEach { consume(it) } }
    awaitResult { consumer.completionHandler(it) }
  }

  private suspend fun consume(reply: Message<T>) {
    runCatching { handle(reply) }
        .onSuccess { handleSuccess(reply, it) }
        .onFailure { handleFailure(reply, it) }
    runCatching { handleFinally(reply) }.onFailure { log.error(it) }
  }

  override suspend fun stop() {
    consumer.unregister().coAwait()
  }
}
